package com.test.es;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategy;
import com.sun.xml.internal.ws.developer.SerializationFeature;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by Think on 2018/4/14.
 */
public class JdbcToEs {

    static ConcurrentLinkedQueue<String> queues = new ConcurrentLinkedQueue<String>();
    static AtomicBoolean isInsert = new AtomicBoolean(true);
    static TransportClient transportClient = null;

    public static void main(String[] args) throws Exception {
        linkEs();
        blukThreadTest();
        WriteData();
    }

    public static void linkEs() throws Exception {
        Settings settings = Settings.builder()
                .put("cluster.name", "ufgoves")
                .build();
        transportClient = new PreBuiltTransportClient(settings);

        TransportAddress address = new InetSocketTransportAddress(InetAddress.getByName("10.10.5.211"), 9300);
        TransportAddress address1 = new InetSocketTransportAddress(InetAddress.getByName("10.10.5.213"), 9300);
        transportClient.addTransportAddress(address);
        transportClient.addTransportAddress(address1);

        String nodeName = transportClient.nodeName();
        System.out.println("nodeName = " + nodeName);
    }

    /**
     * 写入es
     */
    public static void blukThreadTest() {
        final long aa = System.currentTimeMillis();
        final ConcurrentHashMap<String, Boolean> hashMap = new ConcurrentHashMap<String, Boolean>();
        for (int i = 0; i < 20; i++) {
            new Thread(new Runnable() {
                Integer num = 1;

                @Override
                public void run() {
                    hashMap.put(Thread.currentThread().getName(), Boolean.FALSE);
                    final BulkProcessor bulkProcessor = BulkProcessor.builder(transportClient,
                            new BulkProcessor.Listener() {

                                //批量提交之前执行
                                @Override
                                public void beforeBulk(long l, BulkRequest bulkRequest) {

                                }

                                //批量成功后执行
                                @Override
                                public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
                                    System.out.println("l = " + l);
                                    System.out.println("请求数量是 = " + bulkRequest.numberOfActions());
                                    if (bulkResponse.hasFailures()) {
                                        for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
                                            if (itemResponse.isFailed()) {
                                                System.out.println("itemResponse.getFailureMessage() = "
                                                        + itemResponse.getFailureMessage());
                                            }
                                        }
                                    }

                                }

                                //批量失败后执行
                                @Override
                                public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
                                    System.out.println("l = " + l);
                                    System.out.println("happen fail = " +
                                            throwable.getMessage() + " throwable = " + throwable.getCause());

                                }
                            })
                            /*
                            * setBulkActions(1000):每添加1000个request，执行一次bulk操作
                            * setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB)):每达到5M的请求size时，执行一次bulk操作
                            * setFlushInterval(TimeValue.timeValueSeconds(5)):每5s执行一次bulk操作
                            * setConcurrentRequests(1):默认是1，表示积累bulk requests和发送bulk是异步的，其数值表示发送bulk的并发线程数，设置为0表示二者同步的
                            *setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100),3)):当ES由于资源不足发生异常
                            EsRejectedExecutionException重試策略：默认（50ms, 8）,
                            * 策略算法：start + 10 * ((int) Math.exp(0.8d * (currentlyConsumed)) - 1)
                            * */
                            .setBulkActions(10000)
                            .setBulkSize(new ByteSizeValue(100, ByteSizeUnit.MB))
                            .setBackoffPolicy(
                                    BackoffPolicy.exponentialBackoff(
                                            TimeValue.timeValueMillis(100), 3))
                            .setConcurrentRequests(1)
                            .build();
                    while (true) {
                        if (!queues.isEmpty()) {
                            try {
                                String json = queues.poll();
                                if (json == null) continue;
                                String id = UUID.randomUUID().toString();
                                String routing = "abc";
                                //count++;
                                bulkProcessor.add(new IndexRequest("testw", "testt").source(json));
                            } catch (Exception e) {
                                System.out.println("e.getMessage() = " + e.getMessage());
                            }
                        }
                        if (queues.isEmpty() && !isInsert.get()) {
                            bulkProcessor.flush();
                            long jjj = System.currentTimeMillis() - aa;
                            System.out.print("   " + Thread.currentThread().getName()
                                    + ":" + jjj + "   ");
                            hashMap.put(Thread.currentThread().getName(), Boolean.TRUE);
                            while (hashMap.values().contains(Boolean.FALSE)) {
                                try {
                                    Thread.currentThread().sleep(1 * 1000);
                                } catch (Exception e) {
                                    e.printStackTrace(System.out);
                                }
                            }
                            bulkProcessor.close();
                            break;
                        }
                    }

                }
            }).start();
        }
    }

    /**
     * 数据库数据写入queues
     *
     * @throws ClassNotFoundException
     * @throws SQLException
     */
    public static void WriteData() throws ClassNotFoundException, SQLException {
        String tableName = "common_menu";
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        Integer count = 1;
        //List<String> columnName = Arrays.asList("pk_job_history", "log", "stepname", "levels");
        List<String> columnName = Arrays.asList("code", "name", "url", "pk_menu");
        try {
            Class.forName("org.postgresql.Driver");
            String url = "jdbc:postgresql://10.15.0.208:5432/dmp";

            conn = DriverManager.getConnection(url, "dev", "1");

            System.out.println("写入数据开始，成功连接pg：" + tableName);

            String sql = "select * from " + tableName;

            ps = conn.prepareStatement(sql,
                    ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            ps.setFetchSize(Integer.MAX_VALUE);
            rs = ps.executeQuery();


            ResultSetMetaData rsmd = rs.getMetaData();
            int colCount = rsmd.getColumnCount();
            ObjectMapper objectMapper = new ObjectMapper()
                    .setSerializationInclusion(JsonInclude.Include.NON_NULL)
                    .setDateFormat(new SimpleDateFormat("yyyy-MM-dd"))
                    .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
                    //.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)
                    .setPropertyNamingStrategy(
                            PropertyNamingStrategy.CAMEL_CASE_TO_LOWER_CASE_WITH_UNDERSCORES);
            while (rs.next()) {
                Map<String, String> map = new LinkedHashMap<String, String>();
                for (int i = 1; i < colCount; i++) {
                    String name = rsmd.getColumnName(i);
                    if (columnName.contains(name)) {
                        String value = rs.getString(i);
                        map.put(name, value);
                    }
                }
                count++;
                if (map != null && map.size() > 0) {
                    try {
                        queues.add(objectMapper.writeValueAsString(map));
                    } catch (JsonProcessingException e) {
                        System.out.println("e.getMessage() = " + e.getMessage());
                    }
                }
                if (count % 200000 == 0) {
                    int number = queues.size();
                    int jj = number / 200000;
                    System.out.println("index: " + count + ",jj: " + jj + ",number: " + number);
                    while (jj > 0) {//数据量少于200000
//                        try {
//                            Thread.sleep(2000 * jj);//???为什么
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
                        int number2 = queues.size();
                        jj = number2 / 200000;
                        System.out.println("index2: " + count + ",jj: " + jj + ", number2: " + number2);
                    }
                }
            }

            isInsert = new AtomicBoolean(false);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            try {
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        System.out.println(tableName + "数据写入完毕,共有数据：" + --count);
//        try {
//            Thread.sleep(100000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }
}
